package net.fruit.mq.client;

import java.io.IOException;
import java.io.UnsupportedEncodingException;
import java.util.Map;
import java.util.Properties;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;

import lombok.extern.slf4j.Slf4j;
import net.fruit.mq.client.util.MQSer;

/**
 * MQ客户端工具类（生产者/消费者）
 * 
 * @author CNJUN
 *
 */
@Slf4j
public class MQClient {
	private static final String CHARSET = "UTF-8";
    private static Map<String, Producer> _producer = new ConcurrentHashMap<String, Producer>();
	private static Map<String, SendClient> _SendClient = new ConcurrentHashMap<String, SendClient>();
    
	private static Map<String, Consumer> _consumer = new ConcurrentHashMap<String, Consumer>();
    
	static {
		init();
		startProducer();
	}
	
	private static void init() {
	    Set<String> channelNames = MQConfig.stringChannelNames();
	    
	    for (String ch : channelNames) {
	        // producer
	        Properties producerProperties = MQConfig.getProducerProperties(ch);
	        Producer producer = MQFactory.createProducer(producerProperties);
	        _producer.put(ch, producer);
	        _SendClient.put(ch, new SendClient(ch, producer));
	        
	        // consumer
	        Properties consumerProperties = MQConfig.getConsumerProperties(ch);
	        Consumer consumer = MQFactory.createConsumer(consumerProperties);
	        // 订阅的主题
	        Map<String, String> consumerClass = MQConfig.getConsumerClass(ch);
	        for (String topic : consumerClass.keySet()) {
	            String className = consumerClass.get(topic);
	            if (className == null || className.length() == 0) {
	                log.error("not define topic message listener: {}", topic);
	            } else {
    	            try {
                        Class<?> cls = Class.forName(className);
                        if (MessageListener.class.isAssignableFrom(cls)) {
                            Object obj = cls.newInstance();
                            log.info("subscribe topic: {}, {}", topic, className);
                            final MessageListener listener = (MessageListener) obj;
                            consumer.subscribe(topic, "*", new MessageListener() {
                                public Action consume(Message message, ConsumeContext context) {
                                    Action action = listener.consume(message, context);
                                    return action;
                                }
                            });
                        } else {
                            log.error("define topic message listener class is not implements MessageListener.class: {}, {}", topic, className);
                        }
                    } catch (ClassNotFoundException e) {
                        log.error("define topic message listener class not exist: {}, {}", topic, className);
                    } catch (InstantiationException e) {
                        log.error("init topic message listener failure: {}, {}, {}", topic, className, e.getMessage());
                    } catch (IllegalAccessException e) {
                        log.error("init topic message listener failure: {}, {}, {}", topic, className, e.getMessage());
                    }
	            }
	        }
	        _consumer.put(ch, consumer);
	    }
	}
	
	/**
	 * 消息发送客户端
	 * 
	 * @author CNJUN
	 */
	public static class SendClient {
	    private Producer producer;
	    
	    public SendClient(String name, Producer producer) {
	        this.producer = producer;
	    }
	    
	    public synchronized SendResult send(String topic, String tags, String data) {
            try {
                return send(topic, tags, null, data.getBytes(CHARSET), null);
            } catch (UnsupportedEncodingException e) {
                throw new RuntimeException(e);
            }
        }
        
	    public synchronized SendResult send(String topic, String tags, Object data) {
            try {
                return send(topic, tags, null, MQSer.serialize(data), null);
            } catch (IOException e) {
                throw new RuntimeException(e);
            }
        }
	    
	    public synchronized SendResult send(String topic, String tags, byte[] data) {
	        return send(topic, tags, null, data, null);
	    }
	    
	    public synchronized SendResult send(String topic, String tags, String key, byte[] data, Map<String, String> userProperties) {
	        Message msg = new Message(topic, tags, data);
            
            // 设置代表消息的业务关键属性，请尽可能全局唯一。
            // 以方便您在无法正常收到消息情况下，可通过Console查询消息并补发。
            // 注意：不设置也不会影响消息正常收发
            if (key != null && key.length() > 0) {
                msg.setKey(key);
            }
            
            // 延时时间单位为毫秒（ms），指定一个时刻，在这个时刻之后才能被消费，这个例子表示 3秒 后才能被消费
            //long delayTime = 3000;
            //msg.setStartDeliverTime(System.currentTimeMillis() + delayTime);
            
            // 定时消息投递，设置投递的具体时间戳，单位毫秒例如2016-03-07 16:21:00投递
            //long timeStamp =new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").parse("2017-04-17 09:10:00").getTime();
            //msg.setStartDeliverTime(timeStamp);
            
            if (userProperties != null && userProperties.size() > 0) {
                Properties props = new Properties();
                for (Map.Entry<String, String> entry : userProperties.entrySet()) {
                    props.put(entry.getKey(), entry.getValue());
                }
                msg.setUserProperties(props);
            }
            
            return send(msg);
	    }
	    
	    public synchronized SendResult send(Message msg) {
	        //发送消息，只要不抛异常就是成功
            return producer.send(msg);
	    }
	}
	
	public synchronized static SendClient use(String name) {
	    if (!_SendClient.containsKey(name)) {
			throw new RuntimeException("not exist the message service named " + name + ".");
		}
		
		return _SendClient.get(name);
	}
	
	public synchronized static SendClient use() {
	    if (_SendClient.size() == 0) {
	        throw new RuntimeException("not exist the message service.");
	    }
		return _SendClient.get(0);
	}
	
	public static void startProducer() {
	    for (Producer producer : _producer.values()) {
	        if (!producer.isStarted()) {
	            producer.start();
	        }
	    }
    }
	
	public static void shutdownProducer() {
	    for (Producer producer : _producer.values()) {
            if (producer.isStarted()) {
                producer.shutdown();
            }
        }
	}
	
	public static void startConsumer() {
	    for (Consumer consumer : _consumer.values()) {
            if (!consumer.isStarted()) {
                consumer.start();
            }
        }
	}
	
	public static void shutdownConsumer() {
	    for (Consumer consumer : _consumer.values()) {
            if (consumer.isStarted()) {
                consumer.shutdown();
            }
        }
    }
	
}
